package iie.tools;




import com.datastax.driver.core.BoundStatement;
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.ColumnMetadata;
import com.datastax.driver.core.ConsistencyLevel;
import com.datastax.driver.core.DataType;
import static com.datastax.driver.core.DataType.Name.ASCII;
import static com.datastax.driver.core.DataType.Name.BIGINT;
import static com.datastax.driver.core.DataType.Name.BLOB;
import static com.datastax.driver.core.DataType.Name.BOOLEAN;
import static com.datastax.driver.core.DataType.Name.DOUBLE;
import static com.datastax.driver.core.DataType.Name.FLOAT;
import static com.datastax.driver.core.DataType.Name.INT;
import static com.datastax.driver.core.DataType.Name.LIST;
import static com.datastax.driver.core.DataType.Name.MAP;
import static com.datastax.driver.core.DataType.Name.SET;
import static com.datastax.driver.core.DataType.Name.SMALLINT;
import static com.datastax.driver.core.DataType.Name.TEXT;
import static com.datastax.driver.core.DataType.Name.TINYINT;
import static com.datastax.driver.core.DataType.Name.VARCHAR;
import static com.datastax.driver.core.DataType.Name.VARINT;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.QueryOptions;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.ResultSetFuture;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.SocketOptions;
import com.datastax.driver.core.exceptions.InvalidQueryException;
import com.datastax.driver.core.exceptions.NoHostAvailableException;
import com.datastax.driver.core.exceptions.OperationTimedOutException;
import com.datastax.driver.core.exceptions.QueryExecutionException;
import com.datastax.driver.core.policies.DCAwareRoundRobinPolicy;
import com.datastax.driver.core.policies.DefaultRetryPolicy;
import org.apache.commons.codec.DecoderException;
import org.apache.commons.codec.binary.Hex;
import org.codehaus.jackson.map.ObjectMapper;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import java.util.concurrent.ExecutionException;
//import java.util.stream.Collectors;
//import java.util.stream.Collectors; 

/**
 * Created by jinsheng on 16/9/18.
 *
 */
public class Demo {

    private String tableName;
    private String timestampKey;
    private String timestampFieldKey;
    private String filterHasField;
//    private String filterHasNoField;
    private Boolean isUpdate;
    private List<String> updateFields;
    private String updatePrimaryKey;

    private List<ColumnMetadata> columns;
    private Map<String, Type> columnMap = new HashMap<>();
    private final ObjectMapper objectMapper = new ObjectMapper();

    private Cluster cluster;
    private Session session;

    private PreparedStatement insert_statement = null;

    @SuppressWarnings("unused")
    public Demo(String keyspace,
            String dataCenter,
            String coordinatorAddress,
            String userName,
            String password,
            String tableName,
            String timestampKey,
            String timestampFieldKey,
            String filterHasField,
            String filterHasNoField,
            Boolean isUpdate,
            List<String> updateFields,
            String updatePrimaryKey,
            Integer readTimeOut,
            Boolean retry) {
        this.tableName = tableName;
        this.timestampKey = timestampKey;
        this.timestampFieldKey = timestampFieldKey;
        this.filterHasField = filterHasField;
//        this.filterHasNoField = filterHasNoField;
        this.isUpdate = isUpdate;
        this.updateFields = updateFields;
        this.updatePrimaryKey = updatePrimaryKey;
        if (readTimeOut == null) {
            readTimeOut = 12000;
        }

        //�创建链接
        QueryOptions options = new QueryOptions();
        options.setConsistencyLevel(ConsistencyLevel.LOCAL_ONE);
        Cluster.Builder builder = Cluster.builder();
        if (userName.length() > 0) //需要用户验证
        {
            builder
                    .withSocketOptions(new SocketOptions().setConnectTimeoutMillis(120000).setReadTimeoutMillis(readTimeOut))
                    .withLoadBalancingPolicy(DCAwareRoundRobinPolicy.builder().withLocalDc(dataCenter).build())
                    //.withRetryPolicy(DefaultRetryPolicy.INSTANCE)
                    .withCredentials(userName, password).addContactPoint(coordinatorAddress).withQueryOptions(options);
        } else //不需要用户验证
        {
            builder
                    .withSocketOptions(new SocketOptions().setConnectTimeoutMillis(120000).setReadTimeoutMillis(readTimeOut))
                    .withLoadBalancingPolicy(DCAwareRoundRobinPolicy.builder().withLocalDc(dataCenter).build())
                    //.withRetryPolicy(DefaultRetryPolicy.INSTANCE)
                    .addContactPoint(coordinatorAddress).withQueryOptions(options);
        }
        //设置重试策略�
        if (retry == null || retry) {
            this.cluster = builder.withRetryPolicy(DefaultRetryPolicy.INSTANCE).build();
        } else {
            this.cluster = builder.build();
        }
//绑定keyspace
        this.session = this.cluster.connect(keyspace);
        //获取该表元数据Ԫ���
        this.columns = this.cluster.getMetadata().getKeyspace(keyspace).getTable(this.tableName).getColumns();

        //处理每一列的类型并保存，用于更新时确定调用哪个set方法
//        columnMap = columns.parallelStream().collect(Collectors.toMap(ColumnMetadata::getName, col-> resolveType(col.getType())));
//        columnMap = columns.parallelStream().collect(Collectors.toMap(ColumnMetadata::getName, col -> resolveType(col.getType())));

    }

    /**
     * 映射Cassandra类型到{@linkplain Type}枚举
     *
     * @param rawType Cassandra类型对象
     * @return 返回Cassandra类型对应的Type枚举
     */
    private Type resolveType(DataType rawType) {
        switch (rawType.getName()) {
            case ASCII:
            case VARCHAR:
            case TEXT:
                return Type.STRING;
            case TINYINT:
                return Type.TINYINT;
            case SMALLINT:
                return Type.SMALLINT;
            case INT:
                return Type.INT;
            case BIGINT:
            case VARINT:
                return Type.BIGINT;
            case FLOAT:
                return Type.FLOAT;
            case DOUBLE:
                return Type.DOUBLE;
            case BLOB:
                return Type.BLOB;
            case BOOLEAN:
                return Type.BOOLEAN;
            case LIST:
                return Type.LIST;
            case MAP:
                return Type.MAP;
            case SET:
                return Type.SET;
            default:
                throw new IllegalArgumentException("Unsupported type: " + rawType);
        }
    }

    public Session getSession() {
        return this.session;
    }

    /**
     * 设置更新条件
     *
     * @return
     */
    private String getUpdateStr() {
        StringBuilder sb = new StringBuilder();
        // 遍历�?��更新的各个字段并写入where条件�?
        for (String updateField : updateFields) {
            sb.append(updateField)
                    .append("=:")
                    .append(updateField)
                    .append(",");
        }
        sb.deleteCharAt(sb.length() - 1);
        return sb.toString();
    }

    /**
     * 设置更新列的�?
     *
     * @param map 从消息队列中读取的数�?
     * @param bs
     */
    private void setUpdateValue(Map map, BoundStatement bs) {
        // 遍历�?��更新的各个列并分别设置更新�?
        for (String updateField : updateFields) {
            Type type = columnMap.get(updateField);
            Object val = map.get(updateField);
            type.bind(updateField, val, bs);
        }
    }

    /**
     * 发�?数据，传入参数是 List&lt;Map&gt; 类型�?
     *
     * @param dataLists data list
     */
    public void pushData(List<Map<String, Object>> dataLists) throws IOException {

        if (isUpdate) {
            updateData(dataLists);
        } else {
            push_single_json_data(dataLists);
        }
    }

    /**
     * 提交数据到Cassandra
     *
     * @param jsonList 数据列表
     * @return 返回是否提交成功
     * @throws IOException 抛出异常
     */
    private boolean push_single_json_data(List<Map<String, Object>> jsonList) throws IOException {
        if (insert_statement == null) {
            // 预编译insert语句
            insert_statement = session.prepare("INSERT INTO " + this.tableName + " JSON :json if not exists");
        }

        BoundStatement bs;
        //int dropCount = 0;
        List<ResultSetFuture> future = new ArrayList<>();
        for (Object str : jsonList) {

            // 绑定数据并创建BoundStatement对象
            String json = objectMapper.writeValueAsString(str);
            bs = insert_statement.bind(json);
            // 设置记录时间�?
            if (this.timestampKey.length() > 0 && ((Map) str).containsKey(this.timestampKey)) {
                bs.setDefaultTimestamp((Long) ((Map) str).get(this.timestampKey) * 1000000); // second to microsecond
            }            //batchStatement.add(bs);
            // 异步执行查询
            ResultSetFuture result = session.executeAsync(bs);
            future.add(result);
        }

        // 处理语句执行结果
        for (int i = 0; i < future.size(); i++) {
            try {
                // 获取执行结果
                ResultSet rs = future.get(i).get();
                String result_string = rs.toString();
                System.out.println(result_string);
            } catch (OperationTimedOutException | QueryExecutionException | NoHostAvailableException e) {

                // 此处可能由于服务端原因�?成，可稍后重试该操作
            } catch (InterruptedException | ExecutionException | InvalidQueryException e) {
                // 此处有可能是客户端sql语句不正�?
            }
        }

        future.clear();
        return true;
    }

    /**
     * 提交更新数据到Cassandra
     *
     * @param lists 数据列表
     */
    private void updateData(List<Map<String, Object>> lists) {
        if (insert_statement == null) {
            // 预编译update语句
            String sql = "update " + this.tableName + " set "
                    + getUpdateStr() + " where " + this.updatePrimaryKey
                    + "=:key if exists";// IF EXISTS";
            insert_statement = session.prepare(sql);
        }
        List<ResultSetFuture> future = new ArrayList<>();
        for (Object map : lists) {
            Map m = (Map) map;
            //绑定更新数据，并创建BoundStatement对象
            BoundStatement bs = insert_statement.bind();
            setUpdateValue(m, bs);
            // 绑定更新记录的主�?
            columnMap.get(updatePrimaryKey).bind("key", m.get(updatePrimaryKey), bs);
//            
//            columnMap.get("pkey").bind("key", m.get("pkey"), bs);
//            
//            
//            columnMap.get("pkey").bind("pkey", m.get("pkey"), bs);
            
// 设置记录时间�?
            if (this.timestampKey.length() > 0 && m.containsKey(this.timestampKey)) {
                bs.setDefaultTimestamp((Long) m.get(this.timestampKey) * 1000000); // second to microsecond
            }
            future.add(session.executeAsync(bs));
        }
        // 处理执行结果
        for (int i = 0; i < future.size(); i++) {
            try {
                ResultSet rs = future.get(i).get();
                String result_string = rs.toString();
                if (result_string.contains("ERROR")) {
                } else if (rs.wasApplied()) {
                    //更新成功
                } else {
                    // TODO: 2017-03-25 此处可能是由于记录不存在等原因�?成更新未能生�?by linchan
                }

            } catch (Exception e) {
                // 异常捕捉�?30~234�?
            }
        }
        future.clear();
    }

    public void close() {
        this.session.close();
        this.cluster.close();
    }

    public List<ColumnMetadata> getColumns() {
        return columns;
    }

    private enum Type {

        STRING {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else {
                    bs.setString(col, val.toString());
                }
            }
        }, TINYINT {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setByte(col, Long.valueOf(((String) val).toLowerCase()).byteValue());
                } else if (val instanceof Number) {
                    bs.setByte(col, ((Number) val).byteValue());
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, SMALLINT {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setShort(col, Long.valueOf(((String) val).toLowerCase()).shortValue());
                } else if (val instanceof Number) {
                    bs.setShort(col, ((Number) val).shortValue());
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, INT {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setInt(col, Long.valueOf(((String) val).toLowerCase()).intValue());
                } else if (val instanceof Number) {
                    bs.setInt(col, ((Number) val).intValue());
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, BIGINT {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setLong(col, Long.valueOf(((String) val).toLowerCase()));
                } else if (val instanceof Number) {
                    bs.setLong(col, ((Number) val).longValue());
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, FLOAT {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setFloat(col, Long.valueOf(((String) val).toLowerCase()).floatValue());
                } else if (val instanceof Number) {
                    bs.setFloat(col, ((Number) val).floatValue());
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, DOUBLE {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setDouble(col, Long.valueOf(((String) val).toLowerCase()).doubleValue());
                } else if (val instanceof Number) {
                    bs.setDouble(col, ((Number) val).doubleValue());
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, BLOB {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    try {
                        byte[] bytes = Hex.decodeHex(((String) val).toCharArray());
                        bs.setBytes(col, ByteBuffer.wrap(bytes));
                    } catch (DecoderException e) {
                        throw new IllegalArgumentException("wrong value of hex string, column: " + col);
                    }
                } else if (val instanceof byte[]) {
                    bs.setBytes(col, ByteBuffer.wrap((byte[]) val));
                } else if (val instanceof ByteBuffer) {
                    bs.setBytes(col, (ByteBuffer) val);
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, BOOLEAN {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setBool(col, Boolean.valueOf((String) val));
                } else if (val instanceof Boolean) {
                    bs.setBool(col, (Boolean) val);
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, LIST {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof List) {
                    bs.setList(col, (List<?>) val);
                } else if (val instanceof Collection) {
                    List<?> l = new ArrayList<>((Collection<?>) val);
                    bs.setList(col, l);
                } else {
                    List<Object> l = new ArrayList<>();
                    l.add(val);
                    bs.setList(col, l);
                }
            }
        }, MAP {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof Map) {
                    bs.setMap(col, (Map<?, ?>) val);
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, SET {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof Set) {
                    bs.setSet(col, (Set<?>) val);
                } else if (val instanceof Collection) {
                    Set<?> s = new HashSet<>((Collection<?>) val);
                    bs.setSet(col, s);
                } else {
                    Set<Object> s = new HashSet<>();
                    s.add(val);
                    bs.setSet(col, s);
                }
            }
        };

        public void bind(String col, Object val, BoundStatement bs) {
            throw new IllegalArgumentException("wrong type of " + col);
        }
    }

}
